package com.hisense.ovcloud.mqtt.router.service;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.hisense.ovcloud.commons.configs.kafka.ReactiveKafkaSender;
import com.hisense.ovcloud.commons.dto.common.AesInfoInCache;
import com.hisense.ovcloud.commons.dto.common.AesKeyInfoInCache;
import com.hisense.ovcloud.commons.dto.common.mqtt.MqttCtlMsg;
import com.hisense.ovcloud.commons.dto.req.GetAesKeySecretReq;
import com.hisense.ovcloud.commons.dto.req.GetMqttWifiStatsReq;
import com.hisense.ovcloud.commons.dto.req.MqttThirdPartySetCmdReq;
import com.hisense.ovcloud.commons.dto.resp.GetDeviceInfoResp;
import com.hisense.ovcloud.commons.dto.resp.GetMqttWifiStatsResp;
import com.hisense.ovcloud.commons.tools.LocalTimedCacheUtil;
import com.hisense.ovcloud.commons.tools.ZCServerRequestUtil;
import com.hisense.ovcloud.mqtt.router.config.Constants;
import com.hisense.ovcloud.mqtt.router.config.MqttQoS;
import com.hisense.ovcloud.mqtt.router.config.RouterConfig;
import com.hisense.ovcloud.mqtt.router.dto.common.Signer;
import com.hisense.ovcloud.mqtt.router.dto.req.SetCmdReq;
import io.micrometer.common.util.StringUtils;
import io.smallrye.mutiny.tuples.Tuple2;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.HmacUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.Optional;

@Service
@Slf4j
public class CtlService implements InitializingBean {
    private final WebClient webClient;

    private static Scheduler kafkaUploadPool;
    private static Scheduler doubleSendCmdPool;
    private final RouterConfig routerConfig;
    private final ReactiveKafkaSender reactiveKafkaSender;

    public final LocalTimedCacheUtil<AesInfoInCache> aesInfoInCaches;

    @Value("${doubleSendCmd:true}")
    private boolean doubleSendCmd;

    @Value("${defaultMajorDomain:jhl01}")
    private String defaultMajorDomain;

    @Value("${defaultSubDomain:}")
    private String defaultSubDomain;

    @Value("${oldEc2BusinessUrl:http://ac-mqtt-business:5810}")
    private String oldEc2BusinessUrl;

    @Value("${businessUrl:http://ac-mqtt-business:5810}")
    private String businessUrl;

    public CtlService(@Qualifier("base") WebClient webClient, RouterConfig routerConfig, @Qualifier("aesInfoInCache") LocalTimedCacheUtil<AesInfoInCache> aesInfoInCaches, ReactiveKafkaSender reactiveKafkaSender) {
        this.webClient = webClient;
        this.routerConfig = routerConfig;
        this.aesInfoInCaches = aesInfoInCaches;
        this.reactiveKafkaSender = reactiveKafkaSender;
    }

    public Mono<ServerResponse> handleGetDeviceInfo(ServerRequest req) {
        return req.bodyToMono(GetMqttWifiStatsReq.class)
                .flatMap(r -> {
                            if (org.apache.commons.lang3.StringUtils.isNotEmpty(r.getPhysicalId())) {
                                r.setWifiId(r.getPhysicalId());
                            }
                            return webClient.post()
                                    .uri(businessUrl + "/getDeviceInfo")
                                    .body(Mono.just(r), GetMqttWifiStatsReq.class)
                                    .retrieve()
                                    .bodyToMono(GetDeviceInfoResp.class)
                                    .flatMap(res1 -> Mono.just(Tuple2.of(r, res1)));
                        }
                ).zipWhen(tuple -> {
                    if (doubleSendCmd) {
                        return webClient.post()
                                .uri(oldEc2BusinessUrl + "/zc-mqtt-business/v1/getDeviceInfo")
                                .contentType(new MediaType("application", "x-zc-object"))
                                .accept(MediaType.APPLICATION_JSON, new MediaType("application", "x-zc-object"), MediaType.ALL)
                                .bodyValue(JSON.toJSONString(tuple.getItem1()))
                                .retrieve().toEntity(String.class)
                                .onErrorResume(e -> {
                                    log.error("failed to fetch getDeviceInfo from ec2 ac-mqtt-business: {}", e.getMessage());
                                    return Mono.just(ResponseEntity.of(Optional.of(JSONUtil.toJsonStr(GetMqttWifiStatsResp.builder().build()))));
                                })
                                .zipWhen(res3 -> {
                                    if (Objects.requireNonNull(JSON.parseObject(res3.getBody(), GetMqttWifiStatsResp.class)).getOnlineStatus() == Constants.STATUS_ONLINE || tuple.getItem2().getOnlineStatus() == Constants.STATUS_ONLINE) {
                                        tuple.getItem2().setOnlineStatus(Constants.STATUS_ONLINE);
                                    } else {
                                        tuple.getItem2().setOnlineStatus(Constants.STATUS_OFFLINE);
                                    }

                                    return ServerResponse.ok()
                                            .header("X-Zc-Msg-Name", Constants.ZC.Hitachi.HEADER_RESULT_SUCCESS_CODE)
                                            .body(Mono.just(tuple.getItem2()), GetDeviceInfoResp.class);
                                }, (x, y) -> y);
                    } else {
                        return ServerResponse.ok()
                                .header("X-Zc-Msg-Name", Constants.ZC.Hitachi.HEADER_RESULT_SUCCESS_CODE)
                                .body(Mono.just(tuple.getItem2()), GetDeviceInfoResp.class);
                    }

                }, (x, y) -> y);
    }

    public Mono<ServerResponse> handleSetCmd(ServerRequest req) {
        String nonce = ZCServerRequestUtil.getHeader(req, Constants.ZC.Hitachi.HEADER_ZC_Nonce);
        String accessKey = ZCServerRequestUtil.getHeader(req, Constants.ZC.Hitachi.HEADER_ZC_Access_key);
        int timeStamp = Integer.parseInt(ZCServerRequestUtil.getHeader(req, Constants.ZC.Hitachi.HEADER_ZC_Timestamp));
        int timeout = Integer.parseInt(ZCServerRequestUtil.getHeader(req, Constants.ZC.Hitachi.HEADER_ZC_Timeout));
        String signature = ZCServerRequestUtil.getHeader(req, Constants.ZC.Hitachi.HEADER_ZC_Signature);
        int developerId = Integer.parseInt(ZCServerRequestUtil.getHeader(req, Constants.ZC.Hitachi.HEADER_ZC_Developer_id));

        log.info("{}, {}, {} , {} , {}, {}", nonce, accessKey, timeStamp, timeout, signature, developerId);

        // also send cmd to EC2 ac-mqtt-business
        return req.bodyToMono(MqttThirdPartySetCmdReq.class)
                .flatMap(r -> getAesSecret(Mono.just(GetAesKeySecretReq.builder().domain(String.valueOf(r.getMajorDomainId())).aesKey(accessKey).build()))
                        .zipWhen(aesInfoInCache -> {
                            String signatureCalculated = getSignature(
                                    Signer.builder()
                                            .timeout(timeout)
                                            .timestamp(timeStamp)
                                            .nonce(nonce)
                                            .developerId(developerId)
                                            .method("setCmd")
                                            .majorDomain(Optional.ofNullable(aesInfoInCache.getMajorDomain()).orElse(defaultMajorDomain))
                                            .subDomain(defaultSubDomain)
                                            .build(),
                                    aesInfoInCache.getAesSecret()
                            );
                            log.info("signature: {}", signature);
                            log.info("signatureCalculated: {}", signatureCalculated);
                            if (StringUtils.isNotEmpty(aesInfoInCache.getAesSecret()) && signatureCalculated.equals(signature)) {
                                reactiveKafkaSender.sendToKafkaReactiveDirect(routerConfig.getCtlKafkaTopic(), JSONUtil.toJsonStr(
                                        MqttCtlMsg.builder()
                                                .wifiId(r.getPhysicalId())
                                                .majorDomainId(r.getMajorDomainId())
                                                .subDomainId(r.getSubDomainId())
                                                .type(Constants.MQTT.UploadMsgType.TT)
                                                .payload(r.getPayload())
                                                .build()
                                ), kafkaUploadPool);
                                if (doubleSendCmd) {
                                    webClient.post()
                                            .uri(oldEc2BusinessUrl + "/zc-mqtt-business/v1/setCmd")
                                            .contentType(new MediaType("application", "x-zc-object"))
                                            .accept(MediaType.ALL)
                                            .bodyValue(JSON.toJSONString(SetCmdReq.builder()
                                                    .majorDomainId(r.getMajorDomainId())
                                                    .physicalId(r.getPhysicalId())
                                                    .subDomainId(r.getSubDomainId())
                                                    .qos(MqttQoS.AT_LEAST_ONCE.value())
                                                    .payload(r.getPayload())
                                                    .build()))
                                            .retrieve()
                                            .toEntity(String.class)
                                            .onErrorResume(e -> {
                                                log.error("failed to fetch setCmd to ec2 ac-mqtt-business: {}", e.getMessage());
                                                return Mono.just(ResponseEntity.of(Optional.of("failed")));
                                            })
                                            .doOnSuccess(res -> {
                                                log.info("succeed to call setCmd for ec2 ac-mqtt-business: {}, for wifiId {}", oldEc2BusinessUrl, r.getPhysicalId());
                                            })
                                            .doOnError(err -> log.error("failed to call setCmd for ec2 ac-mqtt-business: {}, for wifiId {}, reason : {}", oldEc2BusinessUrl, r.getPhysicalId(), err.getMessage()))
                                            .subscribeOn(doubleSendCmdPool)
                                            .subscribe();
                                }
                                return ServerResponse.ok()
                                        .header("X-Zc-Msg-Name", Constants.ZC.Hitachi.HEADER_RESULT_SUCCESS_CODE)
                                        .body(Mono.just(true), Boolean.class);
                            } else {
                                return ServerResponse.ok()
                                        .header("X-Zc-Msg-Name", Constants.ZC.Hitachi.HEADER_RESULT_FAILED_CODE)
                                        .header("X-Zc-Router-Error", Constants.ZC.Hitachi.HEADER_RESULT_FAILED_ERROR_MESSAGE)
                                        .body(Mono.just(false), Boolean.class);
                            }
                        }, (x, y) -> y));
    }

    public String getSignature(Signer signer, String key) {
        return signature(signer, key);
    }

    public static String signature(Signer signer, String key) {
        String payload = String.format("%d%d%s%d%s%s%s", signer.getTimeout(), signer.getTimestamp(), signer.getNonce(), signer.getDeveloperId(), signer.getMethod(), signer.getMajorDomain(), signer.getSubDomain());
        System.out.println(payload);
        return HmacUtils.hmacSha256Hex(key.getBytes(StandardCharsets.UTF_8), payload.getBytes(StandardCharsets.UTF_8));
    }

    public Mono<AesInfoInCache> getAesSecret(Mono<GetAesKeySecretReq> req) {
        return req.flatMap(r -> Mono.just(Optional.ofNullable(aesInfoInCaches.getFromCache(r.getDomain())).orElse(AesInfoInCache.builder().build()))
                .zipWhen(aesInfoInCache -> {
                    if (StringUtils.isNotEmpty(aesInfoInCache.getAesKey())) {
                        return Mono.just(aesInfoInCache);
                    } else {
                        return webClient.post()
                                .uri(businessUrl + "/getSecretKey")
                                .body(req, GetAesKeySecretReq.class)
                                .retrieve()
                                .bodyToMono(AesKeyInfoInCache.class)
                                .zipWhen(res -> {
                                    if (StringUtils.isNotEmpty(res.getSecret_key())) {
                                        return Mono.just(aesInfoInCaches.saveToCache(r.getDomain(), AesInfoInCache.builder().aesKey(r.getAesKey()).aesSecret(res.getSecret_key()).majorDomain(res.getMajor_domain()).build()))
                                                .zipWhen(x -> Mono.just(aesInfoInCaches.getFromCache(r.getDomain())), (x, y) -> y);
                                    } else {
                                        return Mono.just(AesInfoInCache.builder().build());
                                    }
                                }, (x, y) -> y);
                    }
                }, (x, y) -> y));
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        kafkaUploadPool = Schedulers.newBoundedElastic(routerConfig.getUploadKafkaTopicExecPoolSize(), routerConfig.getUploadKafkaTopicExecQueueSize(), "kafka-upload-pool");
        doubleSendCmdPool = Schedulers.newBoundedElastic(routerConfig.getDoubleSendCmdExecPoolSize(), routerConfig.getDoubleSendCmdExecQueueSize(), "doubleSendCmd-pool");
    }

    public static void main(String[] args) {
        System.out.println(
                signature(
                        Signer.builder()
                                .timestamp(1726283344)
                                .timeout(600)
                                .nonce("xg1r3dpif1x4bwu3")
                                .developerId(3)
                                .method("setCmd")
                                .majorDomain("jhl")
                                .subDomain("")
                                .build(), "1af5681d405ef6ff809fa2832d0f0a5b"
                )
        );
    }
}
